package cn.gemiman.daoyi.module.iot.job;

import cn.gemiman.daoyi.framework.mybatis.core.query.LambdaQueryWrapperX;
import cn.gemiman.daoyi.module.iot.dal.dataobject.kafkamsglogs.KafkaMsgLogsDO;
import cn.gemiman.daoyi.module.iot.dal.mysql.kafkamsglogs.KafkaMsgLogsMapper;
import cn.gemiman.daoyi.module.iot.service.equipmentmsg.EquipmentMsgService;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.text.CharSequenceUtil;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;

@Component
public class KafkaMsgLogsXxlJob {

    @Resource
    private KafkaMsgLogsMapper kafkaMsgLogsMapper;
    @Resource
    private EquipmentMsgService equipmentMsgService;

    @XxlJob("kafkaMsgLogsReadXxlJobHandler")
    public void kafkaMsgLogsReadXxlJobHandler() {
        String params = CharSequenceUtil.emptyToDefault(XxlJobHelper.getJobParam(), "10");
        int pageSize = CharSequenceUtil.isNumeric(params) ? Integer.parseInt(params) : 10;
        XxlJobHelper.log("kafkaMsgLogsReadXxlJobHandler start, pageSize: " + pageSize);
        List<KafkaMsgLogsDO> list = kafkaMsgLogsMapper.selectList(new LambdaQueryWrapperX<KafkaMsgLogsDO>()
                .eq(KafkaMsgLogsDO::getViewed, false)
                .orderByAsc(KafkaMsgLogsDO::getId)
                .last("limit " + pageSize)
        );
        if (CollUtil.isNotEmpty(list)) {
            list.parallelStream().forEach(kafkaMsgLogsDO -> {
                equipmentMsgService.msgHandler(kafkaMsgLogsDO.getCreateTime(), kafkaMsgLogsDO.getMsg());
                kafkaMsgLogsMapper.updateById(new KafkaMsgLogsDO().setId(kafkaMsgLogsDO.getId()).setViewed(true));
            });
        }
        int deleted = kafkaMsgLogsMapper.delete(new LambdaQueryWrapperX<KafkaMsgLogsDO>()
                .eq(KafkaMsgLogsDO::getViewed, true)
                .lt(KafkaMsgLogsDO::getUpdateTime, LocalDateTime.now().minusDays(7L))
        );
        XxlJobHelper.log("kafkaMsgLogsReadXxlJobHandler end, read: " + list.size() + ", deleted: " + deleted);
    }
}
